package storm.bolt;

import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.BasicOutputCollector;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.topology.base.BaseBasicBolt;
import org.apache.storm.tuple.Tuple;
import java.util.Map;

import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisPool;
import redis.clients.jedis.JedisPoolConfig;

public class OrderSaveBolt extends BaseBasicBolt{
    // 初始化Redis连接池
    private static JedisPool pool = new JedisPool(new JedisPoolConfig(), "127.0.0.1");
    // 从连接池中获取一个连接
    private static Jedis jedis = pool.getResource();;
    public void execute(Tuple tuple, BasicOutputCollector collector) {

        String bookname = tuple.getStringByField("book_name");
        bookname = bookname.replace("book_name\":\"","");
        String bookpress = tuple.getStringByField("book_press");
        bookpress = bookpress.replace("book_press\":\"","");
        String booknum = tuple.getStringByField("book_num");
        booknum = booknum.replace("book_num\":\"","");
        String booktype = tuple.getStringByField("book_type");
        booktype = booktype.replace("book_type\":\"","");
        int a= Integer.parseInt(booknum);
        System.out.println(bookname);
        //每接受到一条元组订单数量加一
        jedis.zincrby("bookname",a,bookname);
        //将order的book_press保存在bookpress的有序集合中，每个bookpress元素加一
        jedis.zincrby("bookpress",1,bookpress);
        //将order的book_type保存在booktype的有序集合中，每个booktype元素加一
        jedis.zincrby("booktype",1,booktype);

    }
    public void declareOutputFields(OutputFieldsDeclarer declarer) {

    }
}
